package com.ylz.sjzx.sio.sdk.client.engine;

import com.corundumstudio.socketio.protocol.Packet;
import com.corundumstudio.socketio.protocol.PacketType;
import com.ylz.sjzx.sio.common.core.constant.Const;
import com.ylz.sjzx.sio.common.core.enums.Cmd;
import com.ylz.sjzx.sio.common.mid.cache.redis.JedisTemplate;
import com.ylz.sjzx.sio.sdk.client.core.IMNioBase;
import com.ylz.sjzx.sio.sdk.client.core.ImAllocator;
import com.ylz.sjzx.sio.sdk.model.ChatRoomMessage;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufHolder;
import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.Objects;

/**
 * @author wiqi
 * @version 1.0
 * @date 2020/12/2
 */
public class SocketIoChatEngine extends ChatEngine {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    public SocketIoChatEngine(String chatRoomId) {
        super(chatRoomId);
    }

    private static Object safeDuplicate(Object message) {
        if (message instanceof ByteBuf) {
            return ((ByteBuf) message).retainedDuplicate();
        } else if (message instanceof ByteBufHolder) {
            return ((ByteBufHolder) message).retainedDuplicate();
        } else {
            return ReferenceCountUtil.retain(message);
        }
    }

    /**
     * 聊天室发消息主体功能
     *
     * @param channelGroup
     * @param chatRoomMessage
     */
    @Override
    public void sendMsg(ChannelGroup channelGroup, ChatRoomMessage chatRoomMessage) {
        if (channelGroup == null) {
            return;
        }
        boolean isMsg = Objects.equals(Cmd.getMsg, chatRoomMessage.getCmd());
        //消息结构体
        Packet packet = new Packet(PacketType.MESSAGE);
        packet.setSubType(PacketType.EVENT);
        packet.setName(chatRoomMessage.getCmd().getValue());
        packet.setData(Arrays.asList(chatRoomMessage.getMessage()));
        ByteBuf out = ImAllocator.alloc(isMsg);
        try {
            TextWebSocketFrame textWebSocketFrame = IMNioBase.encoderHandler.imEncodePacket(packet, out);
            if (isMsg) {
                for (Channel next : channelGroup) {
                    //消息刷屏的时候会导致连接断不开
                    boolean isRight = next.isActive() && next.isWritable();
                    if (isRight) {
                        next.writeAndFlush(safeDuplicate(textWebSocketFrame));
                    } else {
                        JedisTemplate.me().incr(Const.PREFIX_YKT + chatRoomMessage.getRoomName() + ":personLoseMessage", chatRoomMessage.getMessageSize());
                    }
                }
            } else {
                //不做水位控制 并发流
                channelGroup.parallelStream().forEach(channel -> {
                    try {
                        channel.writeAndFlush(safeDuplicate(textWebSocketFrame));
                    } catch (Exception e) {
                        logger.error("发送消息给个人失败", e);
                    }
                });
            }
        } catch (Exception e) {
            logger.error("聊天室发送事件操作失败{}", packet, e);
        } finally {
            if (out.refCnt() != 0) {
                ReferenceCountUtil.safeRelease(out);
                if (out.refCnt() != 0) {
                    logger.debug("释放失败");

                }
            }
        }


    }

}
